package com.zhang.second.day07;

import com.zhang.second.utils.OrderEvent;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.util.Collector;

/**
 * @title: 实时对账
 * @author: zhang
 * @date: 2022/2/2 19:25
 */
public class Example4 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<OrderEvent> appZhifu = env
                .addSource(new SourceFunction<OrderEvent>() {
                    @Override
                    public void run(SourceContext<OrderEvent> ctx) throws Exception {
                        OrderEvent e1 = new OrderEvent("order-1", "app-zhifu", 1000L);
                        ctx.collectWithTimestamp(e1, e1.ts);
                        Thread.sleep(1000L);
                        OrderEvent e2 = new OrderEvent("order-2", "app-zhifu", 2000L);
                        ctx.collectWithTimestamp(e2, e2.ts);
                        Thread.sleep(1000L);
                        ctx.emitWatermark(new Watermark(7000L));
                        Thread.sleep(1000L);
                    }

                    @Override
                    public void cancel() {

                    }
                });

        DataStreamSource<OrderEvent> weixinZhifu = env
                .addSource(new SourceFunction<OrderEvent>() {
                    @Override
                    public void run(SourceContext<OrderEvent> ctx) throws Exception {
                        OrderEvent e1 = new OrderEvent("order-1", "weixin-zhifu", 4000L);
                        ctx.collectWithTimestamp(e1, e1.ts);
                        Thread.sleep(1000L);
                        OrderEvent e2 = new OrderEvent("order-2", "weixin-zhifu", 9000L);
                        ctx.collectWithTimestamp(e2, e2.ts);
                        Thread.sleep(1000L);
                        ctx.emitWatermark(new Watermark(7000L));
                        Thread.sleep(1000L);
                    }

                    @Override
                    public void cancel() {

                    }
                });

        appZhifu.keyBy(r -> r.orderId)
                .connect(weixinZhifu.keyBy(r -> r.orderId))
                .process(new CoProcessFunction<OrderEvent, OrderEvent, String>() {
                    private ValueState<OrderEvent> appState;
                    private ValueState<OrderEvent> weixinState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        appState = getRuntimeContext().getState(
                                new ValueStateDescriptor<OrderEvent>(
                                        "app", Types.POJO(OrderEvent.class)
                                )
                        );
                        weixinState = getRuntimeContext().getState(
                                new ValueStateDescriptor<OrderEvent>(
                                        "weixin", Types.POJO(OrderEvent.class)
                                )
                        );
                    }

                    @Override
                    public void processElement1(OrderEvent value, CoProcessFunction<OrderEvent, OrderEvent, String>.Context ctx, Collector<String> out) throws Exception {
                        if (weixinState.value() == null) {
                            //说明微信支付未到达，存入状态 注册定时器 等待5s
                            appState.update(value);
                            ctx.timerService().registerEventTimeTimer(value.ts + 5000L);
                        }else {
                            //微信支付先到达
                            out.collect("订单" + value.orderId + "对账成功，weixin支付先到达，app支付后到达");
                            //清空微信支付
                            weixinState.clear();
                        }
                    }

                    @Override
                    public void processElement2(OrderEvent value, CoProcessFunction<OrderEvent, OrderEvent, String>.Context ctx, Collector<String> out) throws Exception {
                        if (appState.value()==null){
                            //说明qpp支付为到达，存入状态 注册定时器 等待5s
                            weixinState.update(value);
                            ctx.timerService().registerEventTimeTimer(value.ts + 5000L);
                        }else {
                            //app支付先到达
                            out.collect("订单" + value.orderId + "对账成功，app支付先到达，weixin支付后到达");
                            //清空微信支付
                            appState.clear();
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, CoProcessFunction<OrderEvent, OrderEvent, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                        if (appState.value()!=null){
                            out.collect("订单" + appState.value().orderId + "对账失败，weixin支付未到");
                            appState.clear();
                        }

                        if (weixinState.value()!=null){
                            out.collect("订单" + weixinState.value().orderId + "对账失败，app支付未到");
                            weixinState.clear();
                        }
                    }
                })
                .print();

        env.execute();
    }
}
